package com.wjk.kylin.rpc.core.netty.client;

import com.wjk.kylin.rpc.core.netty.codec.MessageCodecSharable;
import com.wjk.kylin.rpc.core.invoker.RpcConsumerFactory;
import com.wjk.kylin.rpc.core.netty.params.PingMessage;
import com.wjk.kylin.rpc.core.serialize.Serializer;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 *
 * @author wjk
 */
public class NettyClientChannelInitializer extends ChannelInitializer<SocketChannel> {
    private static final Logger LOGGER = LoggerFactory.getLogger(NettyClientChannelInitializer.class);

    private final NettyConnectClient nettyConnectClient;

    private static final LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
    private static MessageCodecSharable MESSAGE_CODEC_SHARABLE;
    private static NettyClientHandle NETTY_CLIENT_HANDLE;

    public NettyClientChannelInitializer(RpcConsumerFactory consumerFactory, NettyConnectClient nettyConnectClient, Serializer.Algorithm serializer) {
        this.nettyConnectClient = nettyConnectClient;
        MESSAGE_CODEC_SHARABLE = new MessageCodecSharable(serializer);
        NETTY_CLIENT_HANDLE = new NettyClientHandle(consumerFactory);
    }

    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        //处理粘包，半包 1024kb
        channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1000, 4, 4, 0, 0));
        //打印日志
//        channel.pipeline().addLast(LOGGING_HANDLER);
        //编码解码
        channel.pipeline().addLast(MESSAGE_CODEC_SHARABLE);

        //心跳处理
        //关注读写事件
        channel.pipeline().addLast(new IdleStateHandler(0, 0, 60));
        //处理读写事件
        channel.pipeline().addLast(new ChannelDuplexHandler() {
            @Override
            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                if (evt instanceof IdleStateEvent) {
                    nettyConnectClient.send(new PingMessage());
                    LOGGER.debug(">>>>>>>>>>> kylin-rpc netty client send beat ping");
                } else {
                    super.userEventTriggered(ctx, evt);
                }
            }
        });

        //业务处理
        channel.pipeline().addLast(NETTY_CLIENT_HANDLE);
    }

}
